import org.apache.zookeeper.*;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

import static org.apache.zookeeper.KeeperException.Code;
import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

public class Master implements Watcher {
    enum MasterStates {RUNNING, ELECTED, NOTELECTED};
    private static final Logger LOG = LoggerFactory.getLogger(Master.class);
    private volatile MasterStates state = MasterStates.RUNNING;
    private ZooKeeper zk;
    private String hostPort;
    private Random random = new Random(this.hashCode());
    private String serverId = Integer.toHexString(random.nextInt());

    public Master(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }

    void stopZk() throws InterruptedException {
        zk.close();
    }

    public void process(WatchedEvent watchedEvent) {
        System.out.println(watchedEvent);
    }

    /**
     * 1、我们尝试创建znode节点 /master。如果znode存在，create就会失败。
     * 同时我们想在/master节点的数据字段保存对应这个服务器的唯一ID
     * <p>
     * 2、数据能存储字节缓缓类型的数据，所以去们将int型转换为一个字节数组。
     * 3、ACL策略有OPEN_ACL_UNSAFE(完全开放的权限)，CREATOR_ALL_ACL（为创建者提供身份验证标识的所有权限），
     * READ_ACL_UNSAFE （提供只读权限），AUTH_IDS（），ANYONE_ID_UNSAFE （匿名权限）
     * <p>
     * 4、创建模式 EPHEMERAL_SEQUENTIAL（临时有序 命令-e -s ）EPHEMERAL（临时 -e）
     * PERSISTENT(持久 ) PERSISTENT_SEQUENTIAL(持久有序)
     */
    void runForMaster() {
        //同步调用
//                zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE
//                        , CreateMode.EPHEMERAL);
        //异步调用
        LOG.info("Running for master");
        zk.create("/master", serverId.getBytes(),
                OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                masterCreateCallback, null);
    }

    public boolean isLeader = false;

    void checkMaster() {
//        while (true) {
//            try {
//                Stat stat = new Stat();
//                byte[] data = zk.getData("/master", false, stat);
//
//                isLeader = new String(data).equals(serverId);
//                return true;
//            } catch (NoNodeException e) {
//                return false;
//            } catch (ConnectionLossException e) {
//
//            }
//
//        }
        LOG.info("checkMaster");
        zk.getData("/master", false, masterCheckCallback, null);
    }

    AsyncCallback.DataCallback masterCheckCallback = new AsyncCallback.DataCallback() {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            LOG.info("masterCheckCallback");
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();
                    return;
                case NONODE:
                    runForMaster();
                    return;
            }
        }
    };

    StringCallback masterCreateCallback = new StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            LOG.info("masterCreateCallback");
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();

                    break;
                case OK:
                    state = MasterStates.ELECTED;
//                    takeLeadership();

                    break;
                case NODEEXISTS:
                    state = MasterStates.NOTELECTED;
//                    masterExists();

                    break;
                default:
                    state = MasterStates.NOTELECTED;
                    LOG.error("Something went wrong when running for master.",
                            KeeperException.create(Code.get(rc), path));
            }
            LOG.info("I'm " + (state == MasterStates.ELECTED ? "" : "not ") + "the leader " + serverId);
        }
    };

}